/*
 Xern Entity Relocation Network
 Copyright (C) 2011 Ronie Salgado <roniesalg@gmail.com>

 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal
 in the Software without restriction, including without limitation the rights
 to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 copies of the Software, and to permit persons to whom the Software is
 furnished to do so, subject to the following conditions:

 The above copyright notice and this permission notice shall be included in
 all copies or substantial portions of the Software.

 THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 THE SOFTWARE.
*/

#include <boost/bind.hpp>
#include <fstream>

#include "Xern/Log.hpp"
#include "MessageTemplate.hpp"
#include "MetaverseNode.hpp"

namespace XernMetaverse
{
	const posix_time::time_duration MetaverseNode::NetworkThreadInterval = posix_time::milliseconds(20);
	const posix_time::time_duration MetaverseNode::PacketResendInterval = posix_time::milliseconds(200);
	const posix_time::time_duration MetaverseNode::KeepAliveInterval = posix_time::milliseconds(2000);
	const posix_time::time_duration MetaverseNode::PacketDuplicationTolerance = posix_time::milliseconds(2000);
	const int MetaverseNode::MaxResendCount = 4;

	MetaverseNode::MetaverseNode(bool ipv6)
		: socket(ioService), socketV6(ioService), ipv6(ipv6)
	{
		// Initialize variables.
		inited = false;
		sequenceNumber = 0;
		lastPingId = 0;
		messageTemplatesPath = "media/scripts/message_templates.msg";
		shuttingDown = false;
	}

	MetaverseNode::~MetaverseNode()
	{
	}

	void MetaverseNode::Initialize()
	{
		if(inited)
			return;

		// Make sure the threads don't close immediately.
		shuttingDown = false;

		// Open the socket.
		socket.open(udp::v4());
		if(ipv6)
			socketV6.open(udp::v6());

		// Start the send thread.
		boost::thread st(boost::bind(&MetaverseNode::SendThread, this));
		sendThread.swap(st);

		// Start the receive thread.
		boost::thread rt(boost::bind(&MetaverseNode::ReceiveThread, this));
		receiveThread.swap(rt);

		// Set the inited flag.
		inited = true;
	}

	void MetaverseNode::Shutdown()
	{
		if(!inited)
			return;

		// Send the shutting down message.
		nodeMutex.lock();
		shuttingDown = true;
		ioService.stop();
		nodeMutex.unlock();

		// Wait the child thread to finish.
		sendThread.join();
		receiveThread.join();

		// Unset the inited flag.
		inited = false;
	}

	bool MetaverseNode::ReadTemplates()
	{
		if(templates.templates.size() > 0)
			return true;

		std::ifstream in;
		in.open(messageTemplatesPath.c_str(), std::ios::in);
		if(in.bad() || in.fail())
		{
			LogConsole->Error("Couldn't open message templates: %s", messageTemplatesPath.c_str());
			return false;
		}

		return ParseMessageTemplates(in, templates);
	}

	boost::shared_ptr<Message> MetaverseNode::CreateMessage(const std::string &name)
	{
		return XernMetaverse::CreateMessage(templates, name);
	}
	unsigned int MetaverseNode::GetPacketSequence()
	{
		boost::unique_lock<boost::mutex> l(sequenceMutex);
		return sequenceNumber++;
	}

	void MetaverseNode::SendThread()
	{
		nodeMutex.lock();
		while(!shuttingDown)
		{
			// Don't eat all the cpu.
			nodeMutex.unlock();
			boost::this_thread::sleep(NetworkThreadInterval);
			nodeMutex.lock();

			// Perform actions for each session.
			Sessions::const_iterator it = sessions.begin();
			for(; it != sessions.end(); it++)
			{
				boost::shared_ptr<Session> session = it->second;
				Session_Send(session);
			}
		}

		nodeMutex.unlock();
	}

	void MetaverseNode::Session_Open(boost::shared_ptr<Session> session)
	{
	}

	void MetaverseNode::Session_Send(boost::shared_ptr<Session> session)
	{
		boost::unique_lock<boost::mutex> l(session->sessionMutex);

		switch(session->state)
		{
		default:
		case Session::SS_CONNECTED:
			Session_SendAcks(session);
			Session_KeepAlive(session);
			Session_SendMessages(session);
			Session_ResendPackets(session);
			Session_CleanPackets(session);
			break;
		case Session::SS_DISCONNECTED:
			break;
		}
	}

	void MetaverseNode::Session_SendAcks(boost::shared_ptr<Session> session)
	{
		const int MaxAcksPerPacket = 0xFF;

		// Check the packets to ack.
		int numAcks = session->packetsToAck.size();
		if(numAcks == 0)
			return;

		// Copy the packets to acks.
		std::vector<u32> packetsToAck;
		packetsToAck.reserve(numAcks);
		Session::PacketsToAck::iterator it = session->packetsToAck.begin();
		for(; it != session->packetsToAck.end(); it++)
			packetsToAck.push_back(*it);

		// Clear the session list.
		session->packetsToAck.clear();

		// Calculate the number of packets to send.
		int numPackets = numAcks / MaxAcksPerPacket;
		if(numAcks % MaxAcksPerPacket)
			numPackets++;

		// Create the message.
		boost::shared_ptr<Message> ackMessage = CreateMessage("PacketAck");
		assert(ackMessage);

		MessageBlock &dataBlock = ackMessage->GetBlock("Packets");

		// Create the packet and prepare it.
		Packet ackPacket;
		ackPacket.flags = 0;
		ackPacket.message = ackMessage;

		if(numPackets > 1)
			dataBlock.SetNumberOfSubBlocks(MaxAcksPerPacket);

		// Update the last send time.
		session->lastSentTime = posix_time::microsec_clock::universal_time();

		// Send the acks in different packets.
		int pos = 0;
		int toPut = MaxAcksPerPacket;
		for(int i = 0; i < numPackets; i++)
		{
			if(i + 1 == numPackets)
			{
				toPut = numAcks % MaxAcksPerPacket;
				dataBlock.SetNumberOfSubBlocks(toPut);
			}

			// Fill the message.
			ackPacket.sequenceNumber = GetPacketSequence();

			// Copy the packet sequences.
			for(int j = 0; j < toPut; j++)
				dataBlock.GetVariable("ID", j).SetValue<unsigned int> (packetsToAck[pos + j]);

			// Increase the message position.
			pos += toPut;

			// Encode the packet.
			ackPacket.Encode();

			// Send the packet.
			socket.send_to(boost::asio::buffer(ackPacket.buffer, ackPacket.packetSize), session->endpoint);
		}
	}

	void MetaverseNode::Session_KeepAlive(boost::shared_ptr<Session> session)
	{
		// Check if I have to send a keep-alive.
		posix_time::ptime now = posix_time::microsec_clock::universal_time();
		if(now < session->lastSentTime + KeepAliveInterval)
			return;

		// Find the oldest unacked message.
		unsigned int oldestUnacked = 0;
		posix_time::ptime oldestTime = now;
		Session::PacketsWaiting::const_iterator it = session->packetsWaiting.begin();
		for(; it != session->packetsWaiting.end(); it++)
		{
			boost::shared_ptr<Packet> packet = it->second;
			if(packet->firstSendTime < oldestTime)
			{
				oldestUnacked = it->first;
				oldestTime = packet->firstSendTime;
			}
		}

		// Set the last ping time.
		session->lastPingTime = now;

		// Send a StartPingCheck message.
		{
			// Create the message.
			boost::shared_ptr<Message> message = CreateMessage("StartPingCheck");
			assert(message);
			MessageBlock &agentDataBlock = message->GetBlock("PingID");

			// Set the fields.
			agentDataBlock.GetVariable("PingID").SetValue<u32> (lastPingId++);
			agentDataBlock.GetVariable("OldestUnacked").SetValue<u32> (oldestUnacked);

			// Enqueue the message.
			message->isReliable = true;
			session->sendQueue.push_back(message);
		}
	}

	void MetaverseNode::Session_SendMessages(boost::shared_ptr<Session> session)
	{
		while(!session->sendQueue.empty())
		{
			// Remove the message from the front of the queue.
			boost::shared_ptr<Message> message = session->sendQueue.front();
			session->sendQueue.pop_front();

			// Create the udp packet.
			boost::shared_ptr<Packet> packet(new Packet());
			packet->message = message;
			packet->sequenceNumber = GetPacketSequence();
			message->sequenceNumber = packet->sequenceNumber;

			// Set the packet flags.
			if(message->isReliable)
				packet->flags |= Packet::FL_RELIABLE;

			if(message->GetTemplate()->encoding == MessageTemplate::ME_ZEROCODED)
				packet->flags |= Packet::FL_ZERO_CODE;

			// Encode the packet.
			packet->Encode();

			// Send the packet.
			socket.send_to(boost::asio::buffer(packet->buffer, packet->packetSize), session->endpoint);

			// Store packet send time.
			packet->firstSendTime = posix_time::microsec_clock::universal_time();
			packet->lastSendTime = packet->firstSendTime;

			// Increase the send count and add the packet into ack waiting list.
			if(message->isReliable)
			{
				packet->sendCount++;
				session->packetsWaiting.insert(std::make_pair(packet->sequenceNumber, packet));
			}

			// Update the last send time.
			session->lastSentTime = packet->lastSendTime;
		}
	}

	void MetaverseNode::Session_ResendPackets(boost::shared_ptr<Session> session)
	{
		posix_time::ptime now = posix_time::microsec_clock::universal_time();

		// Check every packet waiting ack last send time.
		Session::PacketsWaiting::iterator it = session->packetsWaiting.begin();
		while(it != session->packetsWaiting.end())
		{
			boost::shared_ptr<Packet> packet = it->second;

			// Check if its time to resend.
			if(now < packet->lastSendTime + PacketResendInterval)
			{
				it++;
				continue;
			}

			// Check the resend count.
			if(packet->sendCount == MaxResendCount)
			{
				// Discard the packet.
				session->packetsWaiting.erase(it++);

				LogConsole->Warning("Discarding network packet.");
				continue;
			}

			// Increase the resend count.
			packet->sendCount++;

			// Set the resent flag.
			packet->flags |= Packet::FL_RESENT;
			packet->buffer[0] |= Packet::FL_RESENT;

			// Resend the packet.
			socket.send_to(boost::asio::buffer(packet->buffer, packet->packetSize), session->endpoint);

			// Update the last resend time.
			now = posix_time::microsec_clock::universal_time();
			packet->lastSendTime = now;

			// Update last session send.
			session->lastSentTime = packet->lastSendTime;

			// Increase the iterator.
			it++;
		}
	}

	void MetaverseNode::Session_CleanPackets(boost::shared_ptr<Session> session)
	{
		posix_time::ptime now = posix_time::microsec_clock::universal_time();
		// Remove the received packet ids after some time to avoid eating
		// all of the memory.
		Session::ReceivedPackets::iterator it = session->receivedPackets.begin();
		while(it != session->receivedPackets.end())
		{
			if(now > it->second + PacketDuplicationTolerance)
				session->receivedPackets.erase(it++);
			else
				it++;
		}
	}

	void MetaverseNode::Session_ReceivePacket(boost::shared_ptr<Session> session, Packet &packet)
	{
		boost::unique_lock<boost::mutex> l(session->sessionMutex);

		// Assume that the packet was already decoded.
		assert(packet.message);

		// Check if we have already received the packet.
		Session::ReceivedPackets::iterator it =
				session->receivedPackets.find(packet.sequenceNumber);
		if(it != session->receivedPackets.end())
		{
			// Check if the packet is reliable.
			if(packet.flags & Packet::FL_RELIABLE)
			{
				// Add the packet to the ack list.
				session->packetsToAck.insert(packet.sequenceNumber);

				// Update the last receive time.
				it->second = posix_time::microsec_clock::universal_time();
			}

			// Ignore the duplicated packet.
			return;
		}

		// Insert the packet into the received map.
		session->receivedPackets.insert(std::make_pair(packet.sequenceNumber, posix_time::microsec_clock::universal_time()));

		// Check the acks flag.
		if(packet.flags & Packet::FL_ACKS)
		{
			for(size_t i = 0; i < packet.acks.size(); i++)
			{
				unsigned int packetId = packet.acks[i];

				// Remove the packet from the waiting list.
				session->packetsWaiting.erase(packetId);
			}
		}

		// Check the reliable flag.
		if(packet.flags & Packet::FL_RELIABLE)
		{
			// Add the packet to the ack list.
			session->packetsToAck.insert(packet.sequenceNumber);
		}

		// Check for connection oriented messages.
		bool store = true;
		boost::shared_ptr<const MessageTemplate> templ =
				packet.message->GetTemplate();
		if(templ->name == "PacketAck")
		{
			// Get the packets block.
			const MessageBlock &packetBlock = packet.message->GetBlock("Packets");
			int numAcks = packetBlock.GetNumberOfSubBlocks();
			for(int i = 0; i < numAcks; i++)
			{
				unsigned int packetId = packetBlock.GetVariable(0, i).GetValue<u32> ();

				// Remove the packet from the waiting list.
				session->packetsWaiting.erase(packetId);
			}
			store = false;
		}
		else if(templ->name == "StartPingCheck")
		{
			// Get the PingID block.
			const MessageBlock &pingBlock = packet.message->GetBlock("PingID");

			// Read the ping id.
			u8 pingId = pingBlock.GetVariable("PingID").GetValue<u32> ();

			// Send a CompletePingCheck message.
			boost::shared_ptr<Message> newMessage = CreateMessage("CompletePingCheck");
			assert(newMessage);

			// Set the fields.
			MessageBlock &rpingBlock = newMessage->GetBlock("PingID");
			rpingBlock.GetVariable("PingID").SetValue<u32> (pingId);

			// Send the message.
			newMessage->isReliable = false;
			session->sendQueue.push_back(newMessage);

			// Don't store the ping check message.
			store = false;
		}
		else if(templ->name == "CompletePingCheck")
		{
			// Don't store the ping response.
			store = false;
		}

		// Also send the message for immediate processing, without filtering.
		if(!OnReceiveMessage(session, packet.message) && store)
		{
			// Store unprocessed messages in a queue.
			session->receiveQueue.push_back(packet.message);
		}
	}

	void MetaverseNode::ReceiveThread()
	{
		// Start receiving.
		socket.async_receive_from(boost::asio::buffer(receivedPacket.buffer), receiveEndpoint,
				boost::bind(&MetaverseNode::OnPacketReceived, this,
						boost::asio::placeholders::error,
						boost::asio::placeholders::bytes_transferred));
		if(ipv6)
			socketV6.async_receive_from(boost::asio::buffer(receivedPacketV6.buffer), receiveEndpointV6,
					boost::bind(&MetaverseNode::OnPacketReceivedV6, this,
							boost::asio::placeholders::error,
							boost::asio::placeholders::bytes_transferred));

		// Start the io service.
		ioService.run();
	}

	void MetaverseNode::OnPacketReceived(const boost::system::error_code &error, size_t size)
	{
		// Lock the node
		boost::unique_lock<boost::mutex> l(nodeMutex);

		if(!error)
		{
			// Create a packet copy
			udp::endpoint ep = receiveEndpoint;
			Packet copy = receivedPacket;
			copy.packetSize = size;

			// Start the next packet reception.
			socket.async_receive_from(boost::asio::buffer(receivedPacket.buffer), receiveEndpoint,
					boost::bind(&MetaverseNode::OnPacketReceived, this,
							boost::asio::placeholders::error,
							boost::asio::placeholders::bytes_transferred));

			// Process the received packet.
			ProcessReceivedPacket(copy, ep);
		}
	}

	void MetaverseNode::OnPacketReceivedV6(const boost::system::error_code &error, size_t size)
	{
		// Lock the node
		boost::unique_lock<boost::mutex> l(nodeMutex);

		if(!error)
		{
			// Create a packet copy
			udp::endpoint ep = receiveEndpointV6;
			Packet copy = receivedPacket;
			copy.packetSize = size;

			// Start the next packet reception.
			socketV6.async_receive_from(boost::asio::buffer(receivedPacketV6.buffer), receiveEndpointV6,
					boost::bind(&MetaverseNode::OnPacketReceivedV6, this,
							boost::asio::placeholders::error,
							boost::asio::placeholders::bytes_transferred));

			// Process the received packet.
			ProcessReceivedPacket(copy, ep);
		}
	}

	void MetaverseNode::ProcessReceivedPacket(Packet &packet, udp::endpoint &endpoint)
	{
		// Decode the packet.
		packet.Decode(templates);

		// Check the message.
		if(!packet.message)
			return;

		// Store the sequence number.
		packet.message->sequenceNumber = packet.sequenceNumber;

		// Send the packet to each one of the session candidates.
		bool processed = false;
		Sessions::const_iterator it = sessions.find(endpoint);
		for(; it != sessions.end(); it++)
		{
			if(it->first != endpoint)
				break;

			Session_ReceivePacket(it->second, packet);
			processed = true;
		}

		// Create a new session if wasn't processed.
		if(!processed)
		{
			boost::shared_ptr<Session> session(new Session());
			session->endpoint = endpoint;
			sessions.insert(std::make_pair(endpoint, session));
			Session_Open(session);
		}

	}


} // namespace XernMetaverse
